package com.ruoyi.common.utils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.function.Supplier;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.CannotAcquireLockException;
import org.springframework.dao.DeadlockLoserDataAccessException;
import org.springframework.data.redis.RedisConnectionFailureException;

/**
 * 线程相关工具类.
 * 
 * @author ruoyi
 */
public class Threads
{
    private static final Logger logger = LoggerFactory.getLogger(Threads.class);
    public static ExecutorService pool;
	static{
		int processors = Runtime.getRuntime().availableProcessors() * 4;
		int size = processors + 1;
		pool = Executors.newFixedThreadPool(size);
		logger.warn("processors {}",size);
	}

    /**
     * sleep等待,单位为毫秒
     */
    public static void sleep(long milliseconds)
    {
        try
        {
            Thread.sleep(milliseconds);
        }
        catch (InterruptedException e)
        {
            return;
        }
    }

    /**
     * 停止线程池
     * 先使用shutdown, 停止接收新任务并尝试完成所有已存在任务.
     * 如果超时, 则调用shutdownNow, 取消在workQueue中Pending的任务,并中断所有阻塞函数.
     * 如果仍然超時，則強制退出.
     * 另对在shutdown时线程本身被调用中断做了处理.
     */
    public static void shutdownAndAwaitTermination(ExecutorService pool)
    {
        if (pool != null && !pool.isShutdown())
        {
            pool.shutdown();
            try
            {
                if (!pool.awaitTermination(120, TimeUnit.SECONDS))
                {
                    pool.shutdownNow();
                    if (!pool.awaitTermination(120, TimeUnit.SECONDS))
                    {
                        logger.info("Pool did not terminate");
                    }
                }
            }
            catch (InterruptedException ie)
            {
                pool.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 打印线程异常信息
     */
    public static void printException(Runnable r, Throwable t)
    {
        if (t == null && r instanceof Future<?>)
        {
            try
            {
                Future<?> future = (Future<?>) r;
                if (future.isDone())
                {
                    future.get();
                }
            }
            catch (CancellationException ce)
            {
                t = ce;
            }
            catch (ExecutionException ee)
            {
                t = ee.getCause();
            }
            catch (InterruptedException ie)
            {
                Thread.currentThread().interrupt();
            }
        }
        if (t != null)
        {
            logger.error(t.getMessage(), t);
        }
    }
    
    
    
    
	/**
	 * 线程池执行任务
	 * @param runnable 任务内容
	 */
	public static void execute(Runnable runnable){
		pool.execute(runnable);
	}
	/**
	 * 线程池执行任务,并将结果返回
	 * @param task 任务内容
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public static Object submit(long expireTime,Callable<?> task) throws InterruptedException, ExecutionException, TimeoutException{
		Future<?> future = pool.submit(task);
		return future.get(expireTime, TimeUnit.MILLISECONDS);
	}


	/**
	 * 对list里的元素并行处理,并待所有任务结束后才停止阻塞
	 * 与jdk8的并行流效果一致
	 * @param list     需要并行处理的元素
	 * @param consumer 消费元素
	 * @param timeout  并行处理超时设置
	 */
	public static <T> void execute(List<T> list, Consumer<T> consumer, int timeout) {
		List<CompletableFuture<Void>> futures = new ArrayList<>();
		for (T t : list) {
			CompletableFuture<Void> f = CompletableFuture.runAsync(() -> {
				synchronized (t) {
					consumer.accept(t);
				}
			}, pool);
			futures.add(f);
		}
		CompletableFuture<Void> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
		try {
			all.get(timeout, TimeUnit.SECONDS);
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	};
	
	public static <T> void execute(List<T> list, Consumer<T> consumer) {
		List<CompletableFuture<Void>> futures = new ArrayList<>();
		for (T t : list) {
			CompletableFuture<Void> f = CompletableFuture.runAsync(() -> {
				synchronized (t) {
					consumer.accept(t);
				}
			}, pool);
			futures.add(f);
		}
		CompletableFuture<Void> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
		// 阻塞等待所有任务执行完成
		all.join();
		System.out.println("所有任务执行完成");
	};
	
	
	
	/**
	 * 尝试多次操作确保操作成功
	 * @param times
	 * @param sleep
	 * @param supplier
	 * @return
	 */
	public static <T> T trydo(int times,int sleep,Supplier<T> supplier) {
		int count = 0;
        //更新过程中会出现死锁,确保更新成功
        while(count < times) {
        	try {
				return supplier.get();
			} catch (Exception e) {
				count++;
				e.printStackTrace();
				sleep(sleep * 1000);
			}
        }
        throw new RuntimeException("try do fail");
	}
	
	/**
	 * 尝试多次操作确保操作成功
	 * @param times
	 * @param sleep
	 * @param supplier
	 * @return
	 */
	public static <T> T trydo(long expireTime,Supplier<T> supplier) {
		int count = 0;
        //Logger logger = DeviceMessageHandler.logger;
        //操作数据库过程中会出现死锁等异常,尝试多次,确保成功
        long start = System.currentTimeMillis();
        long end = start + expireTime;
        Exception temp = null;
        while(start <= end) {
        	try {
				T t = supplier.get();
				if(count > 0) {
		        	//logger.error("try do success {}",System.currentTimeMillis() - startTime);
		        }
				return t;
			} catch (Exception e) {
				count++;
				//e.printStackTrace();
				//如果不是死锁引起的异常,直接退出
				if(e.getClass() != DeadlockLoserDataAccessException.class 
						&& e.getClass() != CannotAcquireLockException.class
						&& e.getClass() != RedisConnectionFailureException.class) {
					//DeviceMessageHandler.logger.error("非死锁异常\n 异常类{} || 异常信息{}",e.getClass(),e.getMessage());
					temp = e;
					break;
				}
				sleep(1000L);
				start = System.currentTimeMillis();
			}
        }
        
        if(start > end) {
        	 throw new RuntimeException("尝试超时");
        }else {
        	temp.printStackTrace();
        	throw new RuntimeException(temp.getMessage());
        }
	}
}
